package com.bi.service.impl;

import com.bi.dto.WideTableDataQueryDTO;
import com.bi.service.ElasticsearchService;
import com.bi.vo.WideTableDataVO;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.PutMappingRequest;
import org.elasticsearch.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.*;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.search.aggregations.metrics.ParsedSingleValueNumericMetricsAggregation;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Elasticsearch服务实现类
 * 
 * @author BI Platform Team
 * @since 1.0.0
 */
@Slf4j
@Service
@SuppressWarnings("deprecation")
public class ElasticsearchServiceImpl implements ElasticsearchService {

    @Autowired
    private RestHighLevelClient elasticsearchClient;

    @Override
    public boolean createIndex(String indexName, Map<String, Object> mapping) {
        try {
            CreateIndexRequest request = new CreateIndexRequest(indexName);
            if (mapping != null && !mapping.isEmpty()) {
                request.mapping(mapping);
            }
            
            CreateIndexResponse response = elasticsearchClient.indices().create(request, RequestOptions.DEFAULT);
            log.info("创建索引成功: {}", indexName);
            return response.isAcknowledged();
        } catch (IOException e) {
            log.error("创建索引失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public boolean deleteIndex(String indexName) {
        try {
            DeleteIndexRequest request = new DeleteIndexRequest(indexName);
            AcknowledgedResponse response = elasticsearchClient.indices().delete(request, RequestOptions.DEFAULT);
            log.info("删除索引成功: {}", indexName);
            return response.isAcknowledged();
        } catch (IOException e) {
            log.error("删除索引失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public boolean indexExists(String indexName) {
        try {
            GetIndexRequest request = new GetIndexRequest(indexName);
            return elasticsearchClient.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("检查索引是否存在失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public boolean bulkInsert(String indexName, List<Map<String, Object>> documents) {
        if (CollectionUtils.isEmpty(documents)) {
            return true;
        }

        try {
            BulkRequest bulkRequest = new BulkRequest();
            
            for (int i = 0; i < documents.size(); i++) {
                Map<String, Object> document = documents.get(i);
                IndexRequest indexRequest = new IndexRequest(indexName)
                    .id(String.valueOf(i + System.currentTimeMillis()))
                    .source(document, XContentType.JSON);
                bulkRequest.add(indexRequest);
            }

            BulkResponse bulkResponse = elasticsearchClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            
            if (bulkResponse.hasFailures()) {
                log.error("批量插入部分失败: {}", bulkResponse.buildFailureMessage());
                return false;
            }
            
            log.info("批量插入成功，索引: {}, 文档数量: {}", indexName, documents.size());
            return true;
        } catch (IOException e) {
            log.error("批量插入失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public boolean updateMapping(String indexName, Map<String, Object> mapping) {
        try {
            PutMappingRequest request = new PutMappingRequest(indexName);
            request.source(mapping);
            
            AcknowledgedResponse response = elasticsearchClient.indices().putMapping(request, RequestOptions.DEFAULT);
            log.info("更新索引映射成功: {}", indexName);
            return response.isAcknowledged();
        } catch (IOException e) {
            log.error("更新索引映射失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public WideTableDataVO queryWideTableData(String indexName, WideTableDataQueryDTO queryDTO) {
        long startTime = System.currentTimeMillis();
        
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder sourceBuilder = buildSearchSourceBuilder(queryDTO);
            searchRequest.source(sourceBuilder);

            SearchResponse searchResponse = elasticsearchClient.search(searchRequest, RequestOptions.DEFAULT);
            
            WideTableDataVO result = new WideTableDataVO();
            result.setTotal(searchResponse.getHits().getTotalHits().value);
            result.setCurrent(queryDTO.getCurrent());
            result.setSize(queryDTO.getSize());
            result.setQueryTime(System.currentTimeMillis() - startTime);

            // 处理搜索结果
            List<Map<String, Object>> records = Arrays.stream(searchResponse.getHits().getHits())
                .map(SearchHit::getSourceAsMap)
                .collect(Collectors.toList());
            result.setRecords(records);

            // 处理聚合结果
            if (searchResponse.getAggregations() != null) {
                result.setAggregations(processAggregations(searchResponse.getAggregations()));
            }

            return result;
        } catch (IOException e) {
            log.error("查询数据失败: {}", indexName, e);
            throw new RuntimeException("查询数据失败", e);
        }
    }

    @Override
    public Map<String, Object> aggregateQuery(String indexName, WideTableDataQueryDTO queryDTO) {
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            
            // 构建查询条件
            BoolQueryBuilder boolQuery = buildBoolQuery(queryDTO.getConditions());
            sourceBuilder.query(boolQuery);
            
            // 构建聚合
            if (queryDTO.getAggregation() != null) {
                buildAggregations(sourceBuilder, queryDTO.getAggregation());
            }
            
            sourceBuilder.size(0); // 不返回文档，只返回聚合结果
            searchRequest.source(sourceBuilder);

            SearchResponse searchResponse = elasticsearchClient.search(searchRequest, RequestOptions.DEFAULT);
            
            if (searchResponse.getAggregations() != null) {
                return processAggregations(searchResponse.getAggregations());
            }
            
            return new HashMap<>();
        } catch (IOException e) {
            log.error("聚合查询失败: {}", indexName, e);
            throw new RuntimeException("聚合查询失败", e);
        }
    }

    @Override
    public WideTableDataVO.FieldStats getFieldStats(String indexName, String fieldName) {
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            
            // 添加统计聚合
            sourceBuilder.aggregation(AggregationBuilders.count("count").field(fieldName))
                         .aggregation(AggregationBuilders.cardinality("distinct_count").field(fieldName))
                         .aggregation(AggregationBuilders.min("min").field(fieldName))
                         .aggregation(AggregationBuilders.max("max").field(fieldName))
                         .aggregation(AggregationBuilders.avg("avg").field(fieldName))
                         .aggregation(AggregationBuilders.sum("sum").field(fieldName));
            
            sourceBuilder.size(0);
            searchRequest.source(sourceBuilder);

            SearchResponse searchResponse = elasticsearchClient.search(searchRequest, RequestOptions.DEFAULT);
            
            WideTableDataVO.FieldStats stats = new WideTableDataVO.FieldStats();
            Aggregations aggregations = searchResponse.getAggregations();
            
            if (aggregations != null) {
                ValueCount count = aggregations.get("count");
                stats.setCount(count.getValue());
                
                Cardinality distinctCount = aggregations.get("distinct_count");
                stats.setDistinctCount(distinctCount.getValue());
                
                Min min = aggregations.get("min");
                if (min.getValue() != Double.POSITIVE_INFINITY) {
                    stats.setMin(min.getValue());
                }
                
                Max max = aggregations.get("max");
                if (max.getValue() != Double.NEGATIVE_INFINITY) {
                    stats.setMax(max.getValue());
                }
                
                Avg avg = aggregations.get("avg");
                if (!Double.isNaN(avg.getValue())) {
                    stats.setAvg(avg.getValue());
                }
                
                Sum sum = aggregations.get("sum");
                stats.setSum(sum.getValue());
            }
            
            return stats;
        } catch (IOException e) {
            log.error("获取字段统计信息失败: {} - {}", indexName, fieldName, e);
            return new WideTableDataVO.FieldStats();
        }
    }

    @Override
    public long getDocumentCount(String indexName) {
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            sourceBuilder.query(QueryBuilders.matchAllQuery());
            sourceBuilder.size(0);
            searchRequest.source(sourceBuilder);

            SearchResponse searchResponse = elasticsearchClient.search(searchRequest, RequestOptions.DEFAULT);
            return searchResponse.getHits().getTotalHits().value;
        } catch (IOException e) {
            log.error("获取文档总数失败: {}", indexName, e);
            return 0L;
        }
    }

    @Override
    public boolean deleteByQuery(String indexName, Map<String, Object> query) {
        // 实现删除逻辑
        // 这里简化实现，实际项目中需要根据具体需求构建删除查询
        return true;
    }

    @Override
    public List<String> suggest(String indexName, String field, String prefix, int size) {
        // 实现搜索建议逻辑
        // 这里简化实现，实际项目中可以使用Elasticsearch的建议API
        return new ArrayList<>();
    }

    @Override
    public void exportData(String indexName, WideTableDataQueryDTO queryDTO, DataExportCallback callback) {
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder sourceBuilder = buildSearchSourceBuilder(queryDTO);
            
            // 使用scroll API进行大数据量导出
            sourceBuilder.size(1000); // 每批次1000条
            searchRequest.source(sourceBuilder);
            searchRequest.scroll(TimeValue.timeValueMinutes(5));

            SearchResponse searchResponse = elasticsearchClient.search(searchRequest, RequestOptions.DEFAULT);
            String scrollId = searchResponse.getScrollId();
            
            do {
                List<Map<String, Object>> batch = Arrays.stream(searchResponse.getHits().getHits())
                    .map(SearchHit::getSourceAsMap)
                    .collect(Collectors.toList());
                
                if (!batch.isEmpty()) {
                    callback.process(batch);
                }
                
                // 继续滚动
                if (scrollId != null) {
                    SearchScrollRequest scrollRequest = new org.elasticsearch.action.search.SearchScrollRequest(scrollId);
                    scrollRequest.scroll(TimeValue.timeValueMinutes(5));
                    searchResponse = elasticsearchClient.scroll(scrollRequest, RequestOptions.DEFAULT);
                    scrollId = searchResponse.getScrollId();
                }
            } while (searchResponse.getHits().getHits().length > 0);
            
        } catch (IOException e) {
            log.error("导出数据失败: {}", indexName, e);
            throw new RuntimeException("导出数据失败", e);
        }
    }

    /**
     * 构建搜索源构建器
     */
    private SearchSourceBuilder buildSearchSourceBuilder(WideTableDataQueryDTO queryDTO) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        
        // 构建查询条件
        BoolQueryBuilder boolQuery = buildBoolQuery(queryDTO.getConditions());
        sourceBuilder.query(boolQuery);
        
        // 设置分页
        int from = (queryDTO.getCurrent() - 1) * queryDTO.getSize();
        sourceBuilder.from(from).size(queryDTO.getSize());
        
        // 设置返回字段
        if (!CollectionUtils.isEmpty(queryDTO.getSelectFields())) {
            sourceBuilder.fetchSource(queryDTO.getSelectFields().toArray(new String[0]), null);
        }
        
        // 设置排序
        if (StringUtils.hasText(queryDTO.getSortField())) {
            SortOrder sortOrder = "asc".equals(queryDTO.getSortOrder()) ? SortOrder.ASC : SortOrder.DESC;
            sourceBuilder.sort(queryDTO.getSortField(), sortOrder);
        }
        
        // 构建聚合
        if (queryDTO.getAggregation() != null) {
            buildAggregations(sourceBuilder, queryDTO.getAggregation());
        }
        
        return sourceBuilder;
    }

    /**
     * 构建布尔查询
     */
    private BoolQueryBuilder buildBoolQuery(List<WideTableDataQueryDTO.QueryCondition> conditions) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        
        if (CollectionUtils.isEmpty(conditions)) {
            boolQuery.must(QueryBuilders.matchAllQuery());
            return boolQuery;
        }
        
        for (WideTableDataQueryDTO.QueryCondition condition : conditions) {
            org.elasticsearch.index.query.QueryBuilder query = buildSingleQuery(condition);
            if ("or".equals(condition.getLogic())) {
                boolQuery.should(query);
            } else {
                boolQuery.must(query);
            }
        }
        
        return boolQuery;
    }

    /**
     * 构建单个查询条件
     */
    private org.elasticsearch.index.query.QueryBuilder buildSingleQuery(WideTableDataQueryDTO.QueryCondition condition) {
        String field = condition.getField();
        String operator = condition.getOperator();
        Object value = condition.getValue();
        
        switch (operator) {
            case "eq":
                return QueryBuilders.termQuery(field, value);
            case "ne":
                return QueryBuilders.boolQuery().mustNot(QueryBuilders.termQuery(field, value));
            case "gt":
                return QueryBuilders.rangeQuery(field).gt(value);
            case "gte":
                return QueryBuilders.rangeQuery(field).gte(value);
            case "lt":
                return QueryBuilders.rangeQuery(field).lt(value);
            case "lte":
                return QueryBuilders.rangeQuery(field).lte(value);
            case "in":
                return QueryBuilders.termsQuery(field, (Collection<?>) value);
            case "nin":
                return QueryBuilders.boolQuery().mustNot(QueryBuilders.termsQuery(field, (Collection<?>) value));
            case "like":
                return QueryBuilders.wildcardQuery(field, "*" + value + "*");
            case "exists":
                return QueryBuilders.existsQuery(field);
            case "range":
                return QueryBuilders.rangeQuery(field)
                    .gte(condition.getStartValue())
                    .lte(condition.getEndValue());
            default:
                return QueryBuilders.termQuery(field, value);
        }
    }

    /**
     * 构建聚合
     */
    private void buildAggregations(SearchSourceBuilder sourceBuilder, WideTableDataQueryDTO.AggregationConfig aggregation) {
        // 分组聚合
        if (!CollectionUtils.isEmpty(aggregation.getGroupByFields())) {
            for (String field : aggregation.getGroupByFields()) {
                sourceBuilder.aggregation(AggregationBuilders.terms(field + "_group").field(field));
            }
        }
        
        // 聚合函数
        if (!CollectionUtils.isEmpty(aggregation.getFunctions())) {
            for (WideTableDataQueryDTO.AggregationFunction function : aggregation.getFunctions()) {
                String aggName = function.getAlias() != null ? function.getAlias() : function.getFunction() + "_" + function.getField();
                
                switch (function.getFunction()) {
                    case "count":
                        sourceBuilder.aggregation(AggregationBuilders.count(aggName).field(function.getField()));
                        break;
                    case "sum":
                        sourceBuilder.aggregation(AggregationBuilders.sum(aggName).field(function.getField()));
                        break;
                    case "avg":
                        sourceBuilder.aggregation(AggregationBuilders.avg(aggName).field(function.getField()));
                        break;
                    case "min":
                        sourceBuilder.aggregation(AggregationBuilders.min(aggName).field(function.getField()));
                        break;
                    case "max":
                        sourceBuilder.aggregation(AggregationBuilders.max(aggName).field(function.getField()));
                        break;
                    case "distinct_count":
                        sourceBuilder.aggregation(AggregationBuilders.cardinality(aggName).field(function.getField()));
                        break;
                }
            }
        }
    }

    /**
     * 处理聚合结果
     */
    private Map<String, Object> processAggregations(Aggregations aggregations) {
        Map<String, Object> result = new HashMap<>();
        
        aggregations.forEach(aggregation -> {
            if (aggregation instanceof Terms) {
                Terms terms = (Terms) aggregation;
                List<Map<String, Object>> buckets = terms.getBuckets().stream()
                    .map(bucket -> {
                        Map<String, Object> bucketMap = new HashMap<>();
                        bucketMap.put("key", bucket.getKey());
                        bucketMap.put("doc_count", bucket.getDocCount());
                        return bucketMap;
                    })
                    .collect(Collectors.toList());
                result.put(aggregation.getName(), buckets);
            } else if (aggregation instanceof ParsedSingleValueNumericMetricsAggregation) {
                ParsedSingleValueNumericMetricsAggregation singleValue = (ParsedSingleValueNumericMetricsAggregation) aggregation;
                result.put(aggregation.getName(), singleValue.value());
            }
        });
        
        return result;
    }
}